package com.google.code.p.restrpc.core.async.mem;

import java.util.Iterator;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;

import com.google.code.p.restrpc.AsyncRpcResult;
import com.google.code.p.restrpc.core.async.AsyncRpcResultCache;
import com.google.code.p.restrpc.core.thread.NamedThreadFactory;

/**
 * @author dz
 */
public class AsyncRpcResultMemCache implements AsyncRpcResultCache {
    
    private ScheduledExecutorService scheduledExecutorService;
    
    private AsyncRpcResultMemCacheOptions options;
    
    private AtomicBoolean started = new AtomicBoolean(false);
    
    private ConcurrentHashMap<String, AsyncRpcResultCacheEntry> entries;
    
    public AsyncRpcResultMemCache(AsyncRpcResultMemCacheOptions options) {
        this.options = options;
        this.scheduledExecutorService = Executors.newScheduledThreadPool(options.getWorker(),
                                                                         new NamedThreadFactory("restrpc-async-result-cache",
                                                                                                true));
        this.entries = new ConcurrentHashMap<String, AsyncRpcResultCacheEntry>();
        start();
    }
    
    /**
     * @return
     */
    public int size() {
        return entries.size();
    }
    
    public boolean isStarted() {
        return this.started.get();
    }
    
    private void start() {
        scheduledExecutorService.scheduleWithFixedDelay(new Runnable() {
            
            public void run() {
                if (entries.isEmpty()) {
                    return;
                }
                
                long idleTimeout = options.getIdleTimeout() * 1000;
                for (Iterator<Map.Entry<String, AsyncRpcResultCacheEntry>> iterator = entries.entrySet()
                                                                                             .iterator(); iterator.hasNext();) {
                    Map.Entry<String, AsyncRpcResultCacheEntry> entry = iterator.next();
                    String id = entry.getKey();
                    AsyncRpcResultCacheEntry cacheEntry = entry.getValue();
                    if ((cacheEntry.getLastMillsStamp() + idleTimeout) < System.currentTimeMillis()) {
                        iterator.remove();
                    }
                }
            }
        },
        options.getDetectFrequency(),
        options.getDetectFrequency(),
        TimeUnit.SECONDS);
        
        started.set(true);
    }
    
    public void shutdown() {
        if (started.get()) {
            this.scheduledExecutorService.shutdown();
            this.started.set(false);
        }
    }
    
    public void clearup() {
        if (!started.get()) {
            throw new MemCacheNotStartedException();
        }
        entries.clear();
    }
    
    public AsyncRpcResult get(String id) {
        if (!started.get()) {
            throw new MemCacheNotStartedException();
        }
        
        AsyncRpcResultCacheEntry cacheEntry = entries.get(id);
        if (cacheEntry == null) {
            return null;
        }
        cacheEntry.updateMillsStamp();
        return cacheEntry.getAsyncRpcResult();
    }
    
    public void put(String id, AsyncRpcResult asyncRpcResult) {
        if (!started.get()) {
            throw new MemCacheNotStartedException();
        }
        
        if (entries.size() + 1 > options.getMaxSize()) {
            throw new MemCacheRejectedException("The cache[maxSize=" + options.getMaxSize()
                                                + "] is full!");
        }
        
        entries.putIfAbsent(id,
                            new AsyncRpcResultCacheEntry(id, asyncRpcResult));
    }
    
    public AsyncRpcResult remove(String id) {
        if (!started.get()) {
            throw new MemCacheNotStartedException();
        }
        
        AsyncRpcResultCacheEntry cacheEntry = entries.remove(id);
        if (cacheEntry == null) {
            return null;
        }
        return cacheEntry.getAsyncRpcResult();
    }
    
    public AsyncRpcResult update(String id, AsyncRpcResult asyncRpcResult) {
        if (!started.get()) {
            throw new MemCacheNotStartedException();
        }
        
        AsyncRpcResultCacheEntry cacheEntry = entries.putIfAbsent(id,
                                                                  new AsyncRpcResultCacheEntry(id,
                                                                                               asyncRpcResult));
        
        if (cacheEntry == null) {
            return null;
        }
        return cacheEntry.getAsyncRpcResult();
    }
    
    class AsyncRpcResultCacheEntry {
        
        private String id;
        
        private AtomicLong lastMillsStamp;
        
        private AsyncRpcResult asyncRpcResult;
        
        AsyncRpcResultCacheEntry(String id, AsyncRpcResult asyncRpcResult) {
            this.id = id;
            this.lastMillsStamp = new AtomicLong(System.currentTimeMillis());
            this.asyncRpcResult = asyncRpcResult;
        }
        
        AsyncRpcResultCacheEntry(String id,
                                 AsyncRpcResult asyncRpcResult,
                                 long lastMillsStamp) {
            this.id = id;
            this.lastMillsStamp = new AtomicLong(lastMillsStamp);
            this.asyncRpcResult = asyncRpcResult;
        }
        
        /**
         * @return the lastMillsStamp
         */
        public long getLastMillsStamp() {
            return lastMillsStamp.get();
        }
        
        /**
         * @return the asyncRpcResult
         */
        public AsyncRpcResult getAsyncRpcResult() {
            return asyncRpcResult;
        }
        
        public void updateMillsStamp() {
            this.lastMillsStamp.getAndSet(System.currentTimeMillis());
        }
        
        public void updateMillsStamp(long lastMillsStamp) {
            this.lastMillsStamp.getAndSet(lastMillsStamp);
        }
        
    }
    
}
